KAFKA-20790: Promote streams assignor API to public module#22788
KAFKA-20790: Promote streams assignor API to public module#22788gabriellefu wants to merge 16 commits into
Conversation
github-actions
Bot
added
triage
gabriellefu
changed the title
gabriellefu
changed the title
mjsax
left a comment
mjsax
left a comment
There was a problem hiding this comment.
Made a pass... Curious to hear from others.
| */ | ||
| @InterfaceAudience.Public | ||
| @InterfaceStability.Evolving | ||
| public record TaskId(String subtopologyId, int partition) implements Comparable<TaskId> { |
There was a problem hiding this comment.
It seems KIP-1357 forget to include this one? Should also become an interface for this case, right?
There was a problem hiding this comment.
I can make it into a interface and update my kip accordingly, thanks
| /** | ||
| * @return The last received cumulative task offsets of assigned tasks or dormant tasks. | ||
| */ | ||
| Map<TaskId, Long> taskOffsets(); |
There was a problem hiding this comment.
One argument during KIP discussion was, to return data based on wire format. That's why we don't have Set<TaskId> activeTasks() but Map<String, Set<Integer>> activeTasks().
And I just realize that the wire-format between both task-ids (which is an array of []TaskId and each TaskId is String,[]int32 format) and task-offsets (which is an array of []TaskOffset and each TaskOffset is string,int32,int64) have a different format...
But I am now wondering what the best representation for string,int32,int64 triple would be -- while Map<TaskId, Long> (with TaskId == String + int) is not a bad fit, it's somewhat a mismatch to what we do for activeTasks() above... 🤔
There was a problem hiding this comment.
I think maybe keep it the Map<TaskId, Long> is more reasonable, I have update the interface of Taskid
| */ | ||
| @InterfaceAudience.Public | ||
| @InterfaceStability.Evolving | ||
| public record GroupAssignment(Map<String, MemberAssignment> members) { |
There was a problem hiding this comment.
The KIP defines this as a class, not a record. -- Given that this is an output/return type, not an input to the assignor, it might actually be ok to keep as record, and we could also update the KIP?
\cc @lianetm for input
There was a problem hiding this comment.
Thanks, I will update my kip and make it into a record.
…get the actual warmuptask
2 participants
Changes
New public API package
org.apache.kafka.coordinator.group.api.assignor.streams(ingroup-coordinator-api), all annotated with@InterfaceAudience.Publicand@InterfaceStability.Evolving.TaskAssignor,GroupAssignment,GroupSpec,MemberAssignment,MemberSubscription,TaskId,TopologyDescriber, andTaskAssignorExceptionfrom the internalstreams.assignorpackage into the new public package.package-info.java.GroupSpecnow exposesmemberIds(),memberSubscription(id),memberAssignmentState(id), andconfigs()instead ofmembers()andassignmentConfigs().MemberSubscriptionis now a thin interface containinginstanceId,rackId,processId, andclientTags.MemberAssignmentStatewithactiveTasks,standbyTasks,warmupTasks,taskOffsets, andtaskEndOffsetsto represent a member’s current assignment state, split out from the previous larger member spec.MemberAssignmentis now an interface exposing onlyactiveTasks()andstandbyTasks(). Warm-up tasks are no longer produced by the assignor and are now determined during reconciliation.Internal implementations
These remain in
group-coordinator, under thestreams.assignorpackage.MemberSubscriptionAndAssignmentImpl, which implements bothMemberSubscriptionandMemberAssignmentState.MemberAssignmentImpl, which implementsMemberAssignment.AssignmentMemberSpecrecord.GroupSpecImplto back the newGroupSpecinterface.StickyTaskAssignor,MockAssignor) andTargetAssignmentBuilderto use the new interfaces.TargetAssignmentBuildernow always emits empty warm-up tasks in the resulting target assignment.GroupMetadataManager,TopologyMetadata,MemberTaskOffsets, andProcessState.Tests and benchmarks
TargetAssignmentBuilderTest,GroupMetadataManagerTest, and the JMH benchmarks (StreamsAssignorBenchmarkUtils,StreamsStickyAssignorBenchmark) to use the new API.GroupSpecImplTestto exercise the new accessors.testAssignmentHasNotChanged,testAssignmentSwapped,testPartialAssignmentUpdate) toACTIVEandSTANDBY, since warm-up tasks can no longer appear in assignor output.Reviewers: Matthias J. Sax matthias@confluent.io